package com.atguigu.userprofile.app

import java.util.Properties

import com.atguigu.userprofile.bean.TagInfo
import com.atguigu.userprofile.constants.CodeConst
import com.atguigu.userprofile.dao.TagInfoDAO
import com.atguigu.userprofile.util.{ClickhouseUtil, MyPropertiesUtil}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object TaskBitmapApp {


  def main(args: Array[String]): Unit = {

   // 因为为了程序调度监控的统一 所以会通过spark-submit 提交到yarn运行  所以必须有sparkcontext
    val sparkConf: SparkConf = new SparkConf().setAppName("task_bitmap_app")//.setMaster("local[*]")
    val sparkContext = new SparkContext(sparkConf)

    var taskId=args(0)
    var taskDate=args(1)

    //  1 组合 查询语句
    //    select   tag.1 tag_code, tag.2 tag_value,groupBitmapState(uid)
    //  from
    //  (
    //    select uid ,
    //    arrayJoin( [('gender',gender),('agegroup',agegroup ),( 'favor',favor).... ] ) tag
    //   from user_tag_merge_2021xxxx
    //   )  tv
    //   group  by tag_code,tag_value

    // ('gender',gender),('agegroup',agegroup ),( 'favor',favor)....
    val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoListWithOn()

    val tagInfoStringListBuffer: ListBuffer[TagInfo]= new ListBuffer[TagInfo]()
    val tagInfoLongListBuffer: ListBuffer[TagInfo]= new ListBuffer[TagInfo]()
    val tagInfoDecimalListBuffer: ListBuffer[TagInfo]= new ListBuffer[TagInfo]()
    val tagInfoDateListBuffer: ListBuffer[TagInfo]= new ListBuffer[TagInfo]()
    //跟据标签值的不同 拆分成不同的集合  准备插到对应的bitmap表中
    for ( tagInfo <- tagInfoList ) {
      if(tagInfo.tagValueType== CodeConst.TAG_VALUE_TYPE_STRING){
        tagInfoStringListBuffer.append(tagInfo)
      }else if (tagInfo.tagValueType== CodeConst.TAG_VALUE_TYPE_LONG){
        tagInfoLongListBuffer.append(tagInfo)
      }else if (tagInfo.tagValueType== CodeConst.TAG_VALUE_TYPE_DECIMAL){
        tagInfoDecimalListBuffer.append(tagInfo)
      }else if (tagInfo.tagValueType== CodeConst.TAG_VALUE_TYPE_DATE){
        tagInfoDateListBuffer.append(tagInfo)
      }
    }

    insertBitmapTable(tagInfoStringListBuffer,taskDate,"user_tag_value_string")
    insertBitmapTable(tagInfoLongListBuffer,taskDate,"user_tag_value_long")
    insertBitmapTable(tagInfoDecimalListBuffer,taskDate,"user_tag_value_decimal")
    insertBitmapTable(tagInfoDateListBuffer,taskDate,"user_tag_value_date")





  }

  //根据不同的标签集合 写入到对应的bitmap表
  def  insertBitmapTable(tagInfoList:ListBuffer[TagInfo] ,taskDate:String ,bitmapTable:String ): Unit ={

    if(tagInfoList.size>0){

    val tagCodeList: List[String] = tagInfoList.toList.map(tagInfo=>  s"('${tagInfo.tagCode}',  ${tagInfo.tagCode.toLowerCase} )" )
    val tagCodeSQL: String = tagCodeList.mkString(",")

    var mergeTableName="user_tag_merge_"+taskDate.replace("-","")

    val properties: Properties = MyPropertiesUtil.load("config.properties")
    val  upDBname: String = properties.getProperty("user-profile.dbname")

    val selectSql=
      s"""
         |      select   tag.1 tag_code, tag.2 tag_value,groupBitmapState( cast ( uid  as  UInt64) ) ,'$taskDate'
         |     from
         |     (
         |      select uid ,
         |       arrayJoin( [ $tagCodeSQL ] ) tag
         |        from $upDBname.$mergeTableName
         |        )  tv
         |        where tag.2 <> ''
         |        group  by tag_code,tag_value
       """.stripMargin

    val  delSQL=s" alter table $bitmapTable delete where dt='$taskDate'  "   //程序中对幂等性的处理
      println(delSQL)
      ClickhouseUtil.executeSql(delSQL)

    val  insertSQL=s" insert into $bitmapTable  $selectSql"

    println(insertSQL)

      ClickhouseUtil.executeSql(insertSQL)
    }
  }

}
